// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: hgpark

#include "coin2/strategy/util/health_reporter_util.h"
#include "coin2/strategy/rmm_strat/rmm_strategy.h"
#include "coin2/strategy/rmm_strat/util/helper.h"

namespace coin2::strategy::rmm_strat {

inline void GetNameMultiplier(const std::string& norm, std::optional<double>& multiplier) {
  if (multiplier.has_value()) return;
  *multiplier = 1.0;
  std::vector<std::string> symbols_1000 = {
    "FLOKI",
    "LUNC",
    "PEPE",
    "PEPE2",
    "SHIB",
    "XEC"
  };
  for (const auto& symbol : symbols_1000) {
    if (norm.find(symbol) != std::string::npos) {
      std::string base_str = norm.substr(0, norm.find("-USD"));
      if (base_str == symbol) {
        *multiplier = 1e3;
      }
    }
  }
}

inline int64_t ToIntKey(double val, double tick_size, int32_t sign) {
  if (sign == -1) {
    return static_cast<int64_t>(std::ceil(val / tick_size - 1e-8));
  } else if (sign == 1) {
    return static_cast<int64_t>(std::floor(val / tick_size + 1e-8));
  } else {
    return static_cast<int64_t>(std::round(val / tick_size));
  }
}

inline double ToFloatKey(int64_t int_price, double tick_size, int32_t sign) {
  if (sign == -1) {
    return static_cast<double>(int_price - 1e-8) * tick_size;
  } else if (sign == 1) {
    return static_cast<double>(int_price + 1e-8) * tick_size;
  } else {
    return int_price * tick_size;
  }
}

std::string GetProductKey(const IProduct* product) {
  return fmt::format("{}.{}", product->absolute_norm(), product->exchange_name());
}

std::string GetProductKey(const OrderEvent* oe) {
  return fmt::format("{}.{}", oe->symbol(), ExchangeTypeToString(oe->exchange_type()));
}

bool HasMidPrice(const FeedUpdate& upd) {
  bool has_midp = false;
  if (upd.book().Bid0().has_value() &&
      upd.book().Ask0().has_value() &&
      upd.book().Bid0().value().price < upd.book().Ask0().value().price) {
    has_midp = true;
  }
  return has_midp;
}

std::optional<double> GetMidPrice(const FeedUpdate& upd) {
  if (HasMidPrice(upd)) {
    return (upd.book().Bid0().value().price + upd.book().Ask0().value().price) / 2;
  } else {
    return std::nullopt;
  }
}

inline coin::proto::AccountRequestProto GetAccRequest(const MarketExchangeApi& mea) {
  coin::proto::AccountRequestProto acc_request;
  acc_request.set_market_type(MarketType_Name(mea.market));
  acc_request.set_exchange(ExchangeType_Name(mea.exchange));
  acc_request.set_api_version(mea.api);
  return acc_request;
}

RmmStrategy::RmmStrategy(const std::experimental::filesystem::path& strat_json, int32_t strat_num,
  bool is_sim, const std::experimental::filesystem::path& sim_result_path, int64_t sim_start_ts)
  :  is_sim_(is_sim),
    strategy_created_time_(GetCurrentTimestamp()),
    strat_num_(strat_num) {
  std::ifstream in(strat_json);
  SPDLOG_INFO("Load strategy json: {}", strat_json.string());
  CHECK_THROW(in.is_open());
  auto strat_config_str = ::nlohmann::json::parse(in).dump();
  RmmStrategyConfig rmm_strat_config;
  auto status = google::protobuf::util::JsonStringToMessage(strat_config_str, &rmm_strat_config);
  CHECK_THROW(status.ok()) << "Fail to parse StratConfig from " << strat_config_str
                           << "\n" << status;
  strat_name_ = rmm_strat_config.strategy_request().strategy_name();
  rmm_strat_config.mutable_strategy_request()->set_strategy_name(strat_name_);
  SPDLOG_INFO(rmm_strat_config.DebugString());

  RiskManagerConfig risk_config;
  risk_config.set_rest_error_window_size_sec(rmm_strat_config.rest_error_window_size_sec());
  risk_manager_ = std::make_unique<RiskManager>(risk_config);

  CHECK_EQ(rmm_strat_config.product_configs().size(), 1) << rmm_strat_config.product_configs().size();
  enable_hedge_ = rmm_strat_config.product_configs()[0].enable_hedge();
  if (enable_hedge_) {
    hedge_strat_name_ = rmm_strat_config.product_configs()[0].hedge_strat_name();
  }
  rmm_sub_strat_ = std::make_unique<RmmSubStrategy>(
      this,
      rmm_strat_config.product_configs()[0],
      strategy_created_time_,
      strat_num,
      is_sim,
      sim_result_path,
      sim_start_ts);
  symbol_list_for_config_service_.push_back(
      rmm_sub_strat_->TradeProduct()->absolute_norm());

  const auto& key = GetProductKey(rmm_sub_strat_->TradeProduct());
  feed_sub_map_[key].push_back(rmm_sub_strat_.get());
  CHECK(order_sub_map_.find(key) == order_sub_map_.end());
  order_sub_map_[key] = rmm_sub_strat_.get();
  if (enable_hedge_) {
    const auto& hedge_key = GetProductKey(rmm_sub_strat_->HedgeProduct());
    order_sub_map_[hedge_key] = rmm_sub_strat_.get();
  }
  rmm_sub_strat_->SetRiskManager(risk_manager_.get());
}

void RmmStrategy::InitOwner(::coin2::app::Driver* driver) {
  const auto& driver_config = driver->config();
  for (const auto& [owner, oe_config] : driver_config.strategy().executors()) {
    if (oe_config.order_system() == "Hedge") {
      CHECK(hedge_owner_.empty()) << hedge_owner_;
      hedge_owner_ = owner;
      hedge_system_name_ = oe_config.order_system();
    } else {
      CHECK(owner_.empty() && system_name_.empty()) << owner_ << " " << system_name_;
      owner_ = owner;
      system_name_ = oe_config.order_system();
    }
  }
  for (const auto& [exchange, order_config] : driver_config.order().exchanges()) {
    if (exchange == hedge_system_name_) {
      const auto& key_filepath = order_config.connection_config().key_filepath();
      auto key = AuthKey::FromFile(key_filepath);
      CHECK(key.owner() == hedge_owner_) <<
        fmt::format("owner not matched: {}, {}", key.owner(), hedge_owner_);
    } else if (exchange == system_name_) {
      const auto& key_filepath = order_config.connection_config().key_filepath();
      auto key = AuthKey::FromFile(key_filepath);
      CHECK(key.owner() == owner_) <<
        fmt::format("owner not matched: {}, {}", key.owner(), owner_);
    }
  }
}

void RmmStrategy::Init(::coin2::app::Driver* driver) {
  SPDLOG_INFO("Rmm Strategy Init.");
  InitOwner(driver);
  strat_reporter_.reset(new StrategyReporter2(
      driver->strategy()->config()));
  if (enable_hedge_) {
    StrategyManagerConfig hedge_strat_manager = driver->strategy()->config();
    hedge_strat_manager.set_strategy_name(hedge_strat_name_);
    hedge_strat_manager.mutable_strategy_health_reporter_config()->clear_from_port();
    hedge_strat_manager.mutable_strategy_health_reporter_config()->clear_to_port();
    hedge_strat_reporter_.reset(new StrategyReporter2(
        hedge_strat_manager));
  }
  rmm_sub_strat_->Init(driver, strat_reporter_.get(), hedge_strat_reporter_.get());
  auto strat_config = driver->strategy()->config();
  CHECK_THROW(strat_name_ == strat_config.strategy_name()) << strat_name_ << " " << strat_config.strategy_name();
  CHECK(!strat_config.config_service_address().empty())
      << strat_config.DebugString();
  config_reporter_.reset(new ConfigReporter(
      strat_config.strategy_name(),
      strat_config.config_service_address(),
      symbol_list_for_config_service_,
      [this](const InternalControlRequest& req) { this->onInternalControlRequest(req); }));
  config_reporter_->Start();
}

void RmmStrategy::InitStrategyReporter2() {
  auto config_for_reserve = new PassiveExecutorConfig();
  double max_position =
    rmm_sub_strat_->OrderExecutor()->GetConfig(*(rmm_sub_strat_->TradeProductHolder())).qty().max_position();
  double min_position =
    rmm_sub_strat_->OrderExecutor()->GetConfig(*(rmm_sub_strat_->TradeProductHolder())).qty().min_position();
  config_for_reserve->set_lot_size(0);
  config_for_reserve->set_max_pos(max_position);
  config_for_reserve->set_min_pos(min_position);
  if (rmm_sub_strat_->Mea().market == MarketType::Spot) {
    config_for_reserve->set_reserve(0.5 * (max_position + min_position));
  }
  auto acc_request = GetAccRequest(rmm_sub_strat_->Mea());
  acc_request.set_owner(owner_);
  SPDLOG_INFO("Trade {}", rmm_sub_strat_->Mea().String());
  strat_reporter_->RegisterExecutorConfig(
      GetCurrentTimestamp(), acc_request,
      *rmm_sub_strat_->TradeProduct(), *config_for_reserve);
  // strat_reporter_->RegisterProduct(*rmm_sub_strat_->TradeProduct());
  if (enable_hedge_) {
    auto config_for_reserve = new PassiveExecutorConfig();
    double max_position =
      rmm_sub_strat_->HedgeOrderExecutor()->GetConfig(*(rmm_sub_strat_->HedgeProductHolder())).qty().max_position();
    double min_position =
      rmm_sub_strat_->HedgeOrderExecutor()->GetConfig(*(rmm_sub_strat_->HedgeProductHolder())).qty().min_position();
    config_for_reserve->set_lot_size(0);
    config_for_reserve->set_max_pos(max_position);
    config_for_reserve->set_min_pos(min_position);
    auto hedge_acc_request = GetAccRequest(rmm_sub_strat_->HedgeMea());
    hedge_acc_request.set_owner(hedge_owner_);
    SPDLOG_INFO("Hedge {}", rmm_sub_strat_->HedgeMea().String());
    hedge_strat_reporter_->RegisterExecutorConfig(
        GetCurrentTimestamp(), hedge_acc_request,
        *rmm_sub_strat_->HedgeProduct(), *config_for_reserve);
    // hedge_strat_reporter_->RegisterProduct(*rmm_sub_strat_->HedgeProduct());
  }
}

bool RmmStrategy::FeedUpToDate(const FeedUpdate& upd) const {
  /* 0: current feed ft - et > 0.4s
   * 1: FeedValid
   */
  int64_t ft = upd.timestamp();
  int64_t et = upd.ts().exchange_publish_timestamp();
  return et == 0 || ft - et < rmm_sub_strat_->ConfigConst().et_ft_diff_valid_ms() * 1'000'000LL;
}

void RmmStrategy::onBookFeed(const FeedUpdate& upd) {
  std::string product_key = GetProductKey(rmm_sub_strat_->TradeProduct());
  CHECK_GT(feed_sub_map_[product_key].size(), 0);
  if (!FeedUpToDate(upd)) {
    SPDLOG_INFO("Feed is not up to date {}, ft {}, et {}", upd.mea().String(),
        upd.timestamp(), upd.ts().exchange_publish_timestamp());
    return;
  }
  for (auto& sub_strat : feed_sub_map_[product_key]) {
    if (!IsPriceOnlySystemName(upd.system_name())) {
      sub_strat->onSnapshotBookFeed(product_key, upd);
    }
    if (IsSnapShotOnlySystemName(upd.system_name())) {
      return;
    }
    sub_strat->onBookFeed(product_key, upd);
  }
  if (order_sub_map_.find(product_key) != order_sub_map_.end()) {
    auto midp = GetMidPrice(upd);
    if (midp.has_value()) {
      if (upd.mea().exchange == ExchangeType::Prex) {
        strat_reporter_->UpdateMidp(*order_sub_map_[product_key]->TradeProduct(), midp);
      } else if (enable_hedge_) {
        std::string hedge_product_key = GetProductKey(rmm_sub_strat_->HedgeProduct());
        hedge_strat_reporter_->UpdateMidp(*order_sub_map_[hedge_product_key]->HedgeProduct(), midp);
      }
    }
  }
}

void RmmStrategy::InitReserveMap() {
}

void RmmStrategy::onTradeFeed(const FeedUpdate& upd) {
  using coin2::exchange::feed::TradeSide;
  std::string product_key = GetProductKey(rmm_sub_strat_->TradeProduct());
  CHECK_GT(feed_sub_map_[product_key].size(), 0);
  if (upd.mea().exchange == ExchangeType::Prex || IsPriceOnlySystemName(upd.system_name())) {
    return;
  }
  if (!FeedUpToDate(upd)) {
    SPDLOG_INFO("Feed is not up to date {}, ft {}, et {}", upd.mea().String(),
      upd.timestamp(), upd.ts().exchange_publish_timestamp());
    return;
  }
  for (auto& sub_strat : feed_sub_map_[product_key]) {
    sub_strat->onTradeFeed(product_key, upd);
  }
}

void RmmStrategy::onIndexFeed(const FeedUpdate& upd) {
  SPDLOG_INFO("{} Index, {}", ExchangeTypeToString(upd.exchange()), upd.index().price);
  std::string product_key = GetProductKey(rmm_sub_strat_->TradeProduct());
  CHECK_GT(feed_sub_map_[product_key].size(), 0);
  for (auto& sub_strat : feed_sub_map_[product_key]) {
    sub_strat->onIndexFeed(product_key, upd);
  }
}

void RmmStrategy::onAccountInfo(const OrderUpdate& upd) {
  if (strat_reporter_ && upd.mea() == rmm_sub_strat_->Mea()) {
    strat_reporter_->onAccountInfo(upd);
  } else if (hedge_strat_reporter_ && upd.mea() == rmm_sub_strat_->HedgeMea() &&
              rmm_sub_strat_->Config().report_hedge_account_info()) {
    hedge_strat_reporter_->onAccountInfo(upd);
  }
  rmm_sub_strat_->onAccountInfo(upd);
}

void RmmStrategy::onAccountOrder(const OrderUpdate& upd) {
  if (upd.system_name() != system_name_ && upd.system_name() != hedge_system_name_) {
    return;
  }
  if (enable_hedge_ && upd.mea() == rmm_sub_strat_->HedgeMea()) {
    const auto& key = GetProductKey(&upd.product_order_info().event());
    if (order_sub_map_.count(key) == 0) {
      SPDLOG_WARN("Received unexpected fills {}", key);
    } else {
      order_sub_map_[key]->onAccountOrder(upd);
    }
  } else if (upd.product_order_info().event().type() == OrderEvent::ORDER_FILLED ||
      upd.product_order_info().event().type() == OrderEvent::ORDER_AUTO_CANCELED ||
      upd.product_order_info().event().type() == OrderEvent::CONVERSION) {
    const auto& key = GetProductKey(&upd.product_order_info().event());
    if (upd.product_order_info().event().type() == OrderEvent::CONVERSION) {
      SPDLOG_INFO("Received conversion info! {}", upd.product_order_info().event().DebugString());
      SPDLOG_INFO("OrderContext: {}", upd.product_order_info().order_contexts().size());
      return;
    }
    if (order_sub_map_.count(key) == 0) {
      SPDLOG_WARN("Received unexpected fills {}", key);
    } else {
      order_sub_map_[key]->onAccountOrder(upd);
    }
  }
}

void RmmStrategy::onOrderLog(const coin::proto::OrderGatewayLog& order_log) {
  if (order_log.account_request().owner() == hedge_owner_) {
    if (order_log.type() == coin::proto::OrderGatewayLog::POSITION) {
      for (const auto& pos : order_log.position().each_position()) {
        if (pos.native_symbol() == rmm_sub_strat_->HedgeProduct()->order_native_symbol()) {
          rmm_sub_strat_->SetHedgePosition(pos.net_position(), order_log.position().query_ts());
        }
      }
    }
    if (hedge_strat_reporter_) {
      hedge_strat_reporter_->GetMutableStrategyLogger()->WriteOrderLog(
          order_log,
          true,
          GetCurrentTimestamp());
    }
    return;
  } else if (order_log.account_request().owner() == owner_) {
    // report order_log to strat_reporter_ only if owner matches
    bool orderlog_chosen = (
        order_log.type() == coin::proto::OrderGatewayLog::POSITION
        || order_log.event().type() == OrderEvent::ORDER_SUBMITTED
        || order_log.event().type() == OrderEvent::ORDER_FILLED);
    if (rmm_sub_strat_->Mea().market == MarketType::Spot) {
      orderlog_chosen |= (order_log.type() == coin::proto::OrderGatewayLog::BALANCE);
    }
    if (orderlog_chosen && strat_reporter_) {
      if (order_log.event().type() == OrderEvent::ORDER_SUBMITTED &&
          order_log.event().event_time() - last_strat_submit_report_time_ > 180 * 1'000'000'000LL) {
        last_strat_submit_report_time_ = order_log.event().event_time();
      } else if (order_log.event().type() == OrderEvent::ORDER_SUBMITTED) {
        return;
      }
      strat_reporter_->GetMutableStrategyLogger()->WriteOrderLog(
          order_log,
          true,
          GetCurrentTimestamp());
    }
  } else {
    // if the order log is from other strat, update other_strat_ts to order log's timestamp.
    if (order_log.event().type() == OrderEvent::ORDER_SUBMITTED) {
      other_strat_ts_ = order_log.event().event_time();
    }
  }
}

void RmmStrategy::onOrderSubsystemError(const OrderUpdate& upd) {
  SPDLOG_INFO("OrderSubsystemError: {}", coin::proto::OrderUpdateType_Name(upd.update_type()));
  risk_manager_->HandleOrderUpdate(upd);
}

void RmmStrategy::onCleanup(Driver* driver) {
  std::this_thread::sleep_for(std::chrono::milliseconds(2000));
  // rmm_sub_strat_->CancelOrdersNearBbo(GetCurrentTimestamp());
  rmm_sub_strat_->OrderExecutor()->CancelOrderAll(*rmm_sub_strat_->TradeProductHolder());
  SPDLOG_INFO("Strategy start timestamp: {}", strategy_created_time_);
  SPDLOG_INFO("Strategy end timestamp: {}", GetCurrentTimestamp());
  SPDLOG_INFO("Running time: {}s", (GetCurrentTimestamp() - strategy_created_time_) / 1e9);
}

void RmmStrategy::onInternalControlRequest(const InternalControlRequest& ctrl_req) {
  CHECK(main_ioc_);
  std::string buf;
  auto action = ctrl_req.action();
  if (action == InternalControlRequest::SET) {
    GetConfigResponse req;
    CHECK(req.ParseFromString(ctrl_req.body()));
    switch (req.type()) {
      case ConfigType::RMM_CONFIG: {
        const auto& config_ctrl_read = req.rmm_config();
        std::string pb_str;
        CHECK(config_ctrl_read.SerializeToString(&pb_str));
        if (rmm_sub_strat_->TradeProduct()->absolute_norm() == config_ctrl_read.norm_product()) {
          main_ioc_->post(
              [pb_str, this](){ rmm_sub_strat_->UpdateControlConfig(pb_str); });
        }
        return;
      }
      default: {
        SPDLOG_INFO("unknown GetConfigRequest: {}", req.DebugString());
        return;
      }
    }
  }
}

RmmConfigState::RmmConfigState(
      RmmStrategy* parent,
      const RmmProductConfig& product_config,
      int32_t strat_num)
    : parent_(parent),
      config_const_(product_config) {
  CHECK(parent_);
  config_const_.set_modular_remainder(strat_num);
}

void RmmConfigState::UpdateControlConfig(const std::string& config_ctrl_pb_str) {
  RmmControlConfig config_ctrl;
  CHECK(config_ctrl.ParseFromString(config_ctrl_pb_str)) << config_ctrl_pb_str;
  SPDLOG_INFO(config_ctrl.DebugString());
  // requirement: oe_ init
  if (!oe_ || !quote_pos_updated) return;
  CHECK_EQ(config_const_.trade_product().norm(), config_ctrl.norm_product());
  config_ = RmmProductConfig();
  config_->CopyFrom(config_const_);

  CHECK(config_const_.has_book_volume_multiplier());
  config_->set_book_volume_multiplier(
      config_const_.book_volume_multiplier() *
      config_ctrl.book_volume_multiplier_mult());

  CHECK(config_const_.has_trade_volume_multiplier());
  config_->set_trade_volume_multiplier(
      config_const_.trade_volume_multiplier() *
      config_ctrl.trade_volume_multiplier_mult());

  CHECK(config_const_.oe_config().common().qty().has_min_order_quote());
  config_->mutable_oe_config()->mutable_common()->mutable_qty()->set_min_order_qty(
      config_const_.oe_config().common().qty().min_order_qty() *
      config_ctrl.min_order_qty_mult());

  CHECK(config_const_.has_num_sample());
  config_->set_num_sample(
      config_const_.num_sample() *
      config_ctrl.num_sample_mult());

  CHECK(config_const_.has_book_update_period_ns());
  config_->set_book_update_period_ns(
      config_const_.book_update_period_ns() *
      config_ctrl.book_update_period_ns_mult());
  CHECK(config_->has_trade_update_period_ns());

  if (risk_manager_->num_of_rest_errors() >= 0) {
    config_->set_book_update_period_ns(
        config_->book_update_period_ns() *
        (1.0 + risk_manager_->num_of_rest_errors() *
        config_->update_period_multiplier_by_timeout()));
    config_->set_trade_update_period_ns(
        config_->trade_update_period_ns() *
        (1.0 + risk_manager_->num_of_rest_errors() *
        config_->update_period_multiplier_by_timeout()));
  }

  CHECK(config_const_.has_lean_by_pos_bps());
  config_->set_lean_by_pos_bps(
      config_const_.lean_by_pos_bps() +
      config_ctrl.lean_by_pos_bps_add());

  CHECK(config_->has_exponential_dist_lambda());
  if (config_ctrl.exponential_dist_lambda_override() != 0) {
    config_->set_exponential_dist_lambda(config_ctrl.exponential_dist_lambda_override());
  }

  CHECK(config_->has_skip_too_outside_pct());
  if (config_ctrl.skip_too_outside_pct_override() != 0) {
    config_->set_skip_too_outside_pct(config_ctrl.skip_too_outside_pct_override());
  }

  CHECK(config_->has_cancel_too_outside_pct());
  if (config_ctrl.cancel_too_outside_pct_override() != 0) {
    config_->set_cancel_too_outside_pct(config_ctrl.cancel_too_outside_pct_override());
  }

  CHECK(config_const_.has_ask0_price_offset_bps());
  CHECK_GE(config_const_.ask0_price_offset_bps(), 0);
  if (config_const_.ask0_price_offset_bps() + config_ctrl.ask0_price_offset_bps_add() >= 0) {
    config_->set_ask0_price_offset_bps(
        config_const_.ask0_price_offset_bps() +
        config_ctrl.ask0_price_offset_bps_add());
  } else {
    // prevent fat finger. should not reach here
    SPDLOG_WARN("wrong ask0_price_offset_bps_add, check grpc config ({} + {} < 0)",
        config_const_.ask0_price_offset_bps(),
        config_ctrl.ask0_price_offset_bps_add());
    config_->set_ask0_price_offset_bps(config_const_.ask0_price_offset_bps());
  }

  CHECK(config_const_.has_bid0_price_offset_bps());
  CHECK_GE(config_const_.bid0_price_offset_bps(), 0);
  if (config_const_.bid0_price_offset_bps() + config_ctrl.bid0_price_offset_bps_add() >= 0) {
    config_->set_bid0_price_offset_bps(
        config_const_.bid0_price_offset_bps() +
        config_ctrl.bid0_price_offset_bps_add());
  } else {
    // prevent fat finger. should not reach here
    SPDLOG_WARN("wrong bid0_price_offset_bps_add, check grpc config ({} + {} < 0)",
        config_const_.bid0_price_offset_bps(),
        config_ctrl.bid0_price_offset_bps_add());
    config_->set_bid0_price_offset_bps(config_const_.bid0_price_offset_bps());
  }

  config_->set_order_quote_ub(order_quote_ub);
  config_->set_order_qty_ub(order_qty_ub);

  CHECK(config_->has_trade_product());
  CHECK(config_->trade_product().has_norm());
  CHECK(config_->trade_product().has_mea());
  CHECK_GT(config_->ref_products_size(), 0);
  for (int32_t i = 0; i < config_->ref_products_size(); ++i) {
    CHECK(config_->ref_products(i).has_norm());
    CHECK(config_->ref_products(i).has_mea());
  }
  CHECK(config_->has_modular_divisor());
  CHECK(config_->has_modular_remainder());
  CHECK(config_->has_index_price_window_ns());
  CHECK(config_->has_leverage_rate());
  CHECK(config_->has_sticky_midp_bps());
  CHECK(config_->has_modular_remainder());
  CHECK(config_->has_update_period_multiplier_by_timeout());
  CHECK(config_->has_enable_telemetry());
  CHECK(config_->has_taker_threshold_bps());
  CHECK(config_->has_book_feed_valid_sec());
  CHECK(config_->has_trade_feed_valid_sec());
  CHECK(config_->has_index_feed_valid_sec());

  CHECK(trade_product_holder_);

  auto* oe_prod = oe_->GetMutableExecutorSub(*trade_product_holder_);
  oe_prod->UpdateScale(config_ctrl.position_mult());
  oe_prod->GetMutableCancelCoolDownLimiter()->ResizeCoolDown(10'000'000'000L, 1);
  parent_->InitStrategyReporter2();
}

RmmSubStrategy::RmmSubStrategy(RmmStrategy* parent,
                               const RmmProductConfig& product_config,
                               int64_t strategy_created_time,
                               int32_t strat_num,
                               bool is_sim,
                               const std::experimental::filesystem::path& sim_result_path,
                               int64_t sim_start_ts)
  : RmmConfigState(parent, product_config, strat_num),
    is_sim_(is_sim),
    strategy_created_time_(strategy_created_time),
    strat_num_(strat_num),
    sim_start_ts_(sim_start_ts),
    driver_(nullptr),
    strat_logger_(nullptr),
    health_reporter_(nullptr),
    ref_downtime_window_(config_const_. oe_config().common().cancel().order_lifetime_ns()),
    spread_window_(30'000'000'000L),
    rmm_position_window_(product_config.hedge_rmm_position_window_sec() * 1'000'000'000L) {
  if (is_sim_) {
    if (sim_result_path.empty()) {
      CHECK(false) << "Empty simulation result path. You can set simulation \
        result path by --sim_result_filepath=YOUR_PATH";
    }
    std::vector<std::string> columns;
    columns.push_back("ts");
    columns.push_back("mea");
    columns.push_back("price");
    columns.push_back("is_midp");
    price_writer_ = std::make_unique<CsvWriter>(sim_result_path, columns);
  }
  enable_telemetry_ = product_config.enable_telemetry();
  trade_product_ = CreateProductFromNormString(
      MarketExchangeApi::FromString(config_const_.trade_product().mea()),
      config_const_.trade_product().norm(),
      is_sim_ ? sim_start_ts : GetCurrentTimestamp());
  if (config_const_.enable_hedge()) {
    hedge_product_ = CreateProductFromNormString(
        MarketExchangeApi::FromString(config_const_.hedge_product().mea()),
        config_const_.hedge_product().norm(),
        is_sim_ ? sim_start_ts : GetCurrentTimestamp());
  }
}

void RmmSubStrategy::Init(
    ::coin2::app::Driver* driver,
    ::coin2::strategy::util::StrategyReporter2* reporter,
    ::coin2::strategy::util::StrategyReporter2* hedge_reporter) {
  using RequestStatus = coin2::base::request::RequestStatus;

  driver_ = driver;
  strat_logger_ = reporter->GetMutableStrategyLogger();
  health_reporter_ = reporter->GetMutableHealthReporter();

  auto prom_report_func = [this, strategy_name = driver->strategy()->config().strategy_name()](RequestStatus infomap) -> void {
    std::vector<std::pair<prometheus::Labels, double>> res;
    LOG(INFO) << "GetApiEndPointStats: " << infomap.size();
    for (auto& [key, info]: infomap) {
      auto& [method, endpoint] = key;
      for (auto& [code, cnt] : info.http_code) {
        prometheus::Labels labels;
        labels["method"] = method;
        labels["endpoint"] = endpoint;
        labels["http_code"] = std::to_string(code);
        labels["strategy_name"] = strategy_name;
        res.emplace_back(std::make_pair(std::move(labels), cnt));
      }
    }
    health_reporter_->Write(std::move(res));
  };
  os()->RegisterHealthReporter(prom_report_func);

  auto get_product_holder = [this](const coin2::exchange::base::symbology::IProduct& product, int64_t timestamp) {
      return this->driver_->product_holder_manager()->GetProductHolder(product, timestamp);
  };
  trade_product_holder_.reset(get_product_holder(
        *trade_product_,
        is_sim_ ? sim_start_ts_ : strategy_created_time_)->Clone());
  if (config_const_.enable_hedge()) {
    hedge_product_holder_.reset(get_product_holder(
          *hedge_product_,
          is_sim_ ? sim_start_ts_ : strategy_created_time_)->Clone());
  }
  tick_size_ = trade_product_holder_->product_info().tick_price().UpTicksize(0);
  tick_qty_ = trade_product_holder_->product_info().tick_qty().Ceil(0);
  for (int32_t i = 0; i < config_const_.ref_products_size(); ++i) {
    ref_products_.push_back(CreateProductFromNormString(
        MarketExchangeApi::FromString(config_const_.ref_products(i).mea()),
        config_const_.ref_products(i).norm(),
        is_sim_ ? sim_start_ts_ : GetCurrentTimestamp()));
    pricers_.push_back(
      std::make_unique<RmmPricer>(
        this,
        ref_products_.back().get(),
        get_product_holder(
          *ref_products_.back(),
          is_sim_ ? sim_start_ts_ : strategy_created_time_)->Clone(),
        config_const_.ref_products(i).mea(),
        is_sim_));
  }
  for (auto& mea : config_const_.price_only_meas()) {
    price_only_meas_.insert(MarketExchangeApi::FromString(mea));
  }
  feed_keys_.insert(GetProductKey(TradeProduct()));

  for (auto& [owner, oe] : driver->strategy()->order_executors(0)) {
    if (owner == parent_->Owner()) {
      CHECK(!oe_);
      oe_ = oe.get();
    } else if (owner == parent_->HedgeOwner()) {
      CHECK(!hedge_oe_);
      hedge_oe_ = oe.get();
    }
  }
  CHECK(oe_) << "failed to init oe_";
  if (config_const_.enable_hedge()) {
    CHECK(hedge_oe_) << "failed to init hedge_oe_";
  }

  if (enable_telemetry_) {
    SPDLOG_INFO("Set Latency Recorder");
    os()->gateway()->set_latency_recorder(strat_logger_->mutable_latency_recorder());
  }
}

/*
void RmmSubStrategy::CancelOrdersNearBbo(int64_t timestamp) {
  const auto oq = oe_->oq(*trade_product_holder_);
  auto working_orders = oq->GetAllProductWorkingOrders(trade_product_->order_native_symbol());
  for (auto* order : working_orders) {
    if (order->direction() == OrderDirection::DIRECTION_SELL &&
        order->order_price() <= fair_midp_ * (1 + config_->skip_too_outside_pct() * 1e-2)) {
      oe_->CancelOrder(*trade_product_holder_, timestamp, order->proc_order_id());
      SPDLOG_ERROR("cancel {}", order->order_price());
    } else if (order->direction() == OrderDirection::DIRECTION_BUY &&
        order->order_price() >= fair_midp_ * (1 - config_->skip_too_outside_pct() * 1e-2)) {
      oe_->CancelOrder(*trade_product_holder_, timestamp, order->proc_order_id());
      SPDLOG_ERROR("cancel {}", order->order_price());
    }
  }
}
*/

void RmmSubStrategy::onBookFeed(const std::string& product_key, const FeedUpdate& upd) {
  auto curr_ts = upd.book().Timestamp() ? upd.book().Timestamp() : upd.timestamp();
  auto start_ts = GetCurrentTimestamp();
  SPDLOG_INFO_EVERY_NS(10'000'000'000L, curr_ts, "OnBookFeed is called.");
  CHECK(feed_keys_.find(product_key) != feed_keys_.end());
  if (enable_telemetry_) {
    auto acc_request = GetAccRequest(upd.mea());
    auto* latency_recorder = strat_logger_->mutable_latency_recorder();
    SetFeedTimeToLatency(
        upd,
        coin::proto::LatencyProto::FEED_LATENCY,
        acc_request,
        true,
        latency_recorder);
    SetFeedTimeToLatency(
        upd,
        coin::proto::LatencyProto::ORDER_LATENCY,
        acc_request,
        false,
        latency_recorder);
    // write health report
    auto strat_req = strat_logger_->GetStrategyRequestProto();
    auto symbol = upd.product().absolute_norm();
    coin2::strategy::util::WriteFeedHealthReport(
        upd, strat_req, acc_request, symbol, health_reporter_);
  }
  if (upd.mea().exchange == ExchangeType::Prex) {
    if (HasMidPrice(upd)) {
      double midp = 0.5 * (
          upd.book().Ask0().value().price +
          upd.book().Bid0().value().price);
      UpdateMidp(upd.timestamp(), midp);
      spread_window_.PushElement(
          curr_ts, upd.book().Ask0().value().price - upd.book().Bid0().value().price);
      spread_window_.UpdateTimeOnly(curr_ts);
      SPDLOG_INFO_EVERY_NS(
          10'000'000'000L,
          curr_ts,
          "Prex Spread ma: {}, raw_bbo: ({}, {})",
          spread_window_.average(),
          upd.book().Ask0().value().price,
          upd.book().Bid0().value().price);
    } else {
      SPDLOG_WARN("NO MIDP: Prex feed");
    }
    return ;
  }

  RmmPricer* pricer = nullptr;
  for (auto& pricer_it : pricers_) {
    if (pricer_it->Mea() == upd.mea()) {
      pricer = pricer_it.get();
      if (pricer->onBookFeed(upd)) {
        ManageQuoteSize(*GetMidPrice(upd) * pricer->NameMultiplier(), curr_ts);
        break;
      } else {
        return;
      }
    }
  }
  // We do not fetch configs from the config server if we runs a rmm as a simulator mode.
  if (is_sim_) {
    config_ = RmmProductConfig();
    config_->CopyFrom(config_const_);
    oe_->GetMutableExecutorSub(*trade_product_holder_)->UpdateScale(1.0);
  }
  if (!config_ || !pricer || !quote_pos_updated ||
      !oe_->HasPosition(*trade_product_holder_) ||
      price_only_meas_.count(upd.mea()) > 0) {
    return;
  }
  if (config_->enable_hedge() &&
      upd.market() == hedge_product_->market() &&
      upd.exchange() == hedge_product_->exchange()) {
    if (HasMidPrice(upd)) {
      hedge_book_ask0_ = upd.book().Ask0().value().price;
      hedge_book_bid0_ = upd.book().Bid0().value().price;
    } else {
      hedge_book_ask0_ = std::nullopt;
      hedge_book_bid0_ = std::nullopt;
    }
  }
  SPDLOG_INFO_EVERY_NS(
      10'000'000'000L,
      curr_ts,
      config_->DebugString());
  if (os()->is_ready() && pricer->BookUpdated() && OrderExecutor()) {
    if (!GetPositionForLeaning(curr_ts)) {
      // skip if we dont't have position
      return;
    }
    if (!CalculateFairMidp(curr_ts, false)) {
      return;
    }
    if (book_ts_ && book_ts_ + config_->book_update_period_ns() > curr_ts) {
      // ignore [t, t + 0.8s] B feed to prevent too frequent update
      return;
    }
    if (!index_ts_ || index_ts_ + config_->index_feed_valid_sec() * 1'000'000'000L < curr_ts) {
      // skip calculation when last prex index feed is not up to date: within [t, t + 60s]
      SPDLOG_WARN("Prex Index is not up-to-date!");
      return;
    }
    if (!AggregateRefOrders(curr_ts)) {
      // skip when no exchange is ready
      return;
    }
    if (risk_manager_->num_of_rest_errors() > 0) {
      SPDLOG_WARN("num_of_rest_errors: {}, window_size: {}",
          risk_manager_->num_of_rest_errors(),
          risk_manager_->rest_stat_window_size());
    }
    SPDLOG_INFO(oe_->DebugString(curr_ts));
    SPDLOG_INFO("Rmm Position: {} {}", trade_product_->absolute_norm(), GetPositionForLeaning(curr_ts).value());
    UpdateWorkingOrders(curr_ts);
    if (book_ts_ + config_->book_update_period_ns() <= curr_ts) {
      double ask0 = aggregated_ask_orders_.begin()->first * tick_size_;
      double bid0 = aggregated_bid_orders_.begin()->first * tick_size_;

      book_ts_ = curr_ts;
      for (int i = 0; i < bid_ocs_.size(); i++) {
        if (i < num_order_) {
          SPDLOG_INFO("[amend] amend called! bid, {}", i);
          FireMakerOrders(bid_ocs_[i]->proc_order_id(), bid0 * (1 - 1 * i * 1e-4), OrderDirection::DIRECTION_BUY, 1, curr_ts);
        } else {
          SPDLOG_INFO("[amend] cancel called! bid, {}", i);
          oe_->CancelOrder(*trade_product_holder_, curr_ts, bid_ocs_[i]->proc_order_id());
        }
      }
      for (int i = 0; i < ask_ocs_.size(); i++) {
        if (i < num_order_) {
          SPDLOG_INFO("[amend] amend called! ask, {}", i);
          FireMakerOrders(ask_ocs_[i]->proc_order_id(), ask0 * (1 + 1 * i * 1e-4), OrderDirection::DIRECTION_SELL, -1, curr_ts);
        } else {
          SPDLOG_INFO("[amend] cancel called! ask, {}", i);
          oe_->CancelOrder(*trade_product_holder_, curr_ts, ask_ocs_[i]->proc_order_id());
        }
      }

      // fire leftover
      for (int i = bid_ocs_.size(); i < num_order_; i++) {
        SPDLOG_INFO("[amend] submit called! bid, {}", i);
        FireMakerOrders(0, bid0 * (1 - 1 * i * 1e-4), OrderDirection::DIRECTION_BUY, 1, curr_ts);
      }
      for (int i = ask_ocs_.size(); i < num_order_; i++) {
        SPDLOG_INFO("[amend] submit called! ask, {}", i);
        FireMakerOrders(0, ask0 * (1 + 1 * i * 1e-4), OrderDirection::DIRECTION_SELL, -1, curr_ts);
      }
      maker_fired_ = true;
    }

    SPDLOG_INFO("Fair Book Spread: {}",
        (aggregated_ask_orders_.begin()->first - aggregated_bid_orders_.begin()->first) * tick_size_);
    SPDLOG_INFO("Ask0: {}({})",
        aggregated_ask_orders_.begin()->first, aggregated_ask_orders_.begin()->second);
    SPDLOG_INFO("Bid0: {}({})",
        aggregated_bid_orders_.begin()->first, aggregated_bid_orders_.begin()->second);
    SPDLOG_INFO("API usage: submit {}, cancel {}",
        oe_->GetSubmitCnt() - submit_cnt_, oe_->GetCancelCnt() - cancel_cnt_);
    submit_cnt_ = oe_->GetSubmitCnt();
    cancel_cnt_ = oe_->GetCancelCnt();
  }
  SPDLOG_INFO("Book execution time: {}s", (GetCurrentTimestamp() - start_ts) / 1e9);
}

void RmmSubStrategy::onSnapshotBookFeed(const std::string& product_key, const FeedUpdate& upd) {
  if (upd.mea().exchange == ExchangeType::Prex) {
    return;
  }
  auto curr_ts = upd.book().Timestamp() ? upd.book().Timestamp() : upd.timestamp();
  for (auto& pricer_it : pricers_) {
    if (pricer_it->Mea() == upd.mea()) {
      pricer_it->UpdateRefOrders(upd, curr_ts);
    }
  }
}

std::optional<double> RmmSubStrategy::GetPosition(AccountType account_type, int64_t curr_ts) {
  auto validate_position = [curr_ts, this]
    (const std::optional<std::pair<double, int64_t>>& pos) -> std::optional<double> {
    if (!pos.has_value()) {
      return std::nullopt;
    }
    if (curr_ts - pos.value().second < this->config_const_.position_valid_sec() * 1'000'000'000LL) {
      return pos.value().first;
    }
    return std::nullopt;
  };
  switch (account_type) {
    case AccountType::RMM_STRAT_0:
      return validate_position(strat0_position_);
    case AccountType::RMM_STRAT_1:
      return validate_position(strat1_position_);
    case AccountType::HEDGE:
      return validate_position(hedge_position_);
    default:
      return std::nullopt;
  }
}

std::optional<double> RmmSubStrategy::GetPositionForLeaning(int64_t curr_ts) {
  if (strat_num_ == 0) {
    if (!GetPosition(AccountType::RMM_STRAT_0, curr_ts).has_value()) {
      return std::nullopt;
    }
    if (config_const_.enable_hedge() == 0) {
      double cur_pos = GetPosition(AccountType::RMM_STRAT_0, curr_ts).value();
      leaning_position_window_.Update(curr_ts, cur_pos);
      return cur_pos - leaning_position_window_.average();
    }
    return GetPosition(AccountType::RMM_STRAT_0, curr_ts).value()
      + GetPosition(AccountType::RMM_STRAT_1, curr_ts).value_or(0)
      + GetPosition(AccountType::HEDGE, curr_ts).value_or(0);
  } else if (strat_num_ == 1) {
    return GetPosition(AccountType::RMM_STRAT_1, curr_ts);
  }
  CHECK(false) << "wrong strat_num " << strat_num_;
}

AccountType RmmSubStrategy::GetAccountType(const AccountInfo& account_info) {
  if (account_info.owner() == parent_->Owner()) {
    if (strat_num_ == 0) {
      return AccountType::RMM_STRAT_0;
    } else {
      return AccountType::RMM_STRAT_1;
    }
  } else if (account_info.owner() == parent_->HedgeOwner()) {
    return AccountType::HEDGE;
  } else {
    if (strat_num_ == 0) {
      return AccountType::RMM_STRAT_1;
    } else {
      return AccountType::RMM_STRAT_0;
    }
  }
  return AccountType::UNKNOWN;
}

double RmmSubStrategy::LeanByPosition(double position) {
  double max_position = oe_->GetConfig(*trade_product_holder_).qty().max_position();
  // TODO(peter) : refactor this
  position = position - std::clamp(position, -hedge_exception_qty, hedge_exception_qty);
  CHECK_THROW(max_position > 0) << "Check max_position: " << max_position;
  if (position >= 0) {
    return -(std::pow(position / max_position, 0.6)) * config_->lean_by_pos_bps();
  } else {
    return (std::pow(-position / max_position, 0.6)) * config_->lean_by_pos_bps();
  }
}

void RmmSubStrategy::WritePriceToCsv(int64_t timestamp, std::string mea, double price, bool is_midp) {
  std::vector<std::string> row;
  row.push_back(std::to_string(timestamp));
  row.push_back(mea);
  row.push_back(std::to_string(price));
  row.push_back(std::to_string(is_midp));
  price_writer_->WriteRow(row);
}

bool RmmSubStrategy::CalculateFairMidp(int64_t timestamp, bool enable_logging) {
  /*
   * calculate midp of Prex book
   */

  int32_t price_only_ready_cnt = 0;

  std::vector<double> weighted_fair_prices;
  std::vector<double> ref_weights;

  for (auto& pricer_it : pricers_) {
    bool ref_ready =
      pricer_it->SpotRef() ?
      timestamp - pricer_it->BookTimestamp() < config_->book_feed_valid_sec() * 1'000'000'000LL :
      pricer_it->FairPriceWindow()->Ready(
          timestamp,
          pricer_it->BookTimestamp(),
          config_->index_feed_valid_sec() * 1'000'000'000LL /* index feed valid ns (60s) */,
          config_->book_feed_valid_sec() * 1'000'000'000LL /* book feed valid ns (30s) */);
    if (pricer_it->BookUpdated() && ref_ready) {
      double ref_weight = pricer_it->RefWeight();
      if (pricer_it->SpotRef()) {
        pricer_it->SetFairPrice(pricer_it->Midp() * pricer_it->NameMultiplier());
        weighted_fair_prices.push_back(pricer_it->FairPrice() * ref_weight);
      } else {
        pricer_it->SetFairPrice(
            pricer_it->FairPriceWindow()->GetSelfFairPrice(
              pricer_it->Midp() * pricer_it->NameMultiplier()));
        weighted_fair_prices.push_back(pricer_it->FairPrice() * ref_weight);
        if (enable_logging) {
          SPDLOG_INFO("Prex Index MA: {}, {} Index MA: {}, MA size: {}",
              pricer_it->FairPriceWindow()->self_indexp_window.average(),
              ExchangeTypeToString(pricer_it->Mea().exchange),
              pricer_it->FairPriceWindow()->ref_indexp_window.average(),
              pricer_it->FairPriceWindow()->ref_indexp_window.num_elements());
        }
      }
      ref_weights.push_back(pricer_it->RefWeight());
      if (enable_logging) {
        SPDLOG_INFO("{} ready, Mid Price: {}, Fair Price: {}",
            pricer_it->Mea().String(),
            pricer_it->Midp(),
            pricer_it->FairPrice());
      }
      if (price_only_meas_.count(pricer_it->Mea()) > 0) {
        ++price_only_ready_cnt;
      }
      pricer_it->SetReady(true);
    } else {
      if (enable_logging) {
        SPDLOG_INFO("{} not ready", pricer_it->Mea().String());
      }
      pricer_it->SetReady(false);
    }
  }
  if (enable_logging) {
    SPDLOG_INFO("# of ready exchanges: {}", weighted_fair_prices.size());
  }
  if (weighted_fair_prices.size() == 0 || (price_only_ready_cnt == weighted_fair_prices.size()) ||
      std::accumulate(ref_weights.begin(), ref_weights.end(), 0.0) <= 0.0) {
    SPDLOG_WARN(
        "weighted_fair_prices is empty, size: {}, total_weight {}",
        weighted_fair_prices.size(),
        std::accumulate(ref_weights.begin(), ref_weights.end(), 0.0)
        );
    return false;
  }
  double raw_fair_midp_wo_offset =
    std::accumulate(weighted_fair_prices.begin(), weighted_fair_prices.end(), 0.0) /
    std::accumulate(ref_weights.begin(), ref_weights.end(), 0.0);
  std::vector<double> weighted_fair_price_offsets;
  for (auto& pricer_it : pricers_) {
    if (pricer_it->Ready()) {
      pricer_it->FairPriceOffsetWindow()->Update(timestamp, raw_fair_midp_wo_offset - pricer_it->FairPrice());
      weighted_fair_price_offsets.push_back(pricer_it->FairPriceOffsetWindow()->average() * pricer_it->RefWeight());
      if (enable_logging) {
        SPDLOG_INFO("{} Fair Price Offset: {}, MA size: {}",
            pricer_it->Mea().String(),
            pricer_it->FairPriceOffsetWindow()->average(),
            pricer_it->FairPriceOffsetWindow()->num_elements());
      }
    }
  }
  double fair_price_offset =
    std::accumulate(weighted_fair_price_offsets.begin(), weighted_fair_price_offsets.end(), 0.0) /
    std::accumulate(ref_weights.begin(), ref_weights.end(), 0.0);
  double raw_fair_midp_w_offset = raw_fair_midp_wo_offset + fair_price_offset;
  double lean_bp = LeanByPosition(GetPositionForLeaning(timestamp).value());
  fair_midp_ = raw_fair_midp_w_offset;
  if (lean_bp < 0) {
    fair_ask0_ = raw_fair_midp_w_offset;
    fair_bid0_ = raw_fair_midp_w_offset * (1 + lean_bp * 1e-4);
  } else {
    fair_ask0_ = raw_fair_midp_w_offset * (1 + lean_bp * 1e-4);
    fair_bid0_ = raw_fair_midp_w_offset;
  }
  fair_ask0_ *= (1 + 1e-4 * config_->ask0_price_offset_bps());
  fair_bid0_ *= (1 - 1e-4 * config_->bid0_price_offset_bps());
  int64_t order_lifetime_ns = config_->oe_config().common().cancel().order_lifetime_ns();
  outside_only_ = (strat_num_ == 1 && timestamp - parent_->OtherStratTimestamp() < order_lifetime_ns);
  if (outside_only_) {
    fair_ask0_ *= (1 + 1e-4 * config_->bot1_outside_range_bps());
    fair_bid0_ *= (1 - 1e-4 * config_->bot1_outside_range_bps());
  }
  if (enable_logging) {
    SPDLOG_INFO("FairMidp: {} (={} + {})",
        raw_fair_midp_w_offset,
        raw_fair_midp_wo_offset,
        fair_price_offset);
    SPDLOG_INFO("fair ask0: {}, fair bid0: {}, lean_bp(fair spread): {}",
        fair_ask0_,
        fair_bid0_,
        lean_bp);
  }
  num_sample_ = config_->num_sample();
  return true;
}

bool RmmSubStrategy::AggregateRefOrders(int64_t timestamp) {
  /*
   * make reference exchange's feed fit to Prex's ticksize
   */
  if (!CalculateFairMidp(timestamp, true)) {
    return false;
  }
  if (is_sim_) {
    WritePriceToCsv(timestamp, config_const_.trade_product().mea(), fair_midp_, true/* is_midp */);
    if (book_ts_ + config_->book_update_period_ns() <= timestamp) {
      book_ts_ = timestamp;
    }
    return false;
  }
  aggregated_ask_orders_.clear();
  aggregated_bid_orders_.clear();
  for (auto& pricer_it : pricers_) {
    if (pricer_it->BookUpdated() &&
        timestamp - pricer_it->BookTimestamp() < config_->book_feed_valid_sec() * 1'000'000'000LL &&
        !pricer_it->AskOrders().empty() &&
        !pricer_it->BidOrders().empty()) {
      // aggregate all orders if book feed is up to date
      int64_t snapshot_ask_tick_offset = ToIntKey(fair_ask0_ - pricer_it->SnapshotMidp() * pricer_it->NameMultiplier(), tick_size_, 0);
      int64_t snapshot_bid_tick_offset = ToIntKey(fair_bid0_ - pricer_it->SnapshotMidp() * pricer_it->NameMultiplier(), tick_size_, 0);
      int64_t curr_tick_offset = ToIntKey(fair_midp_ - pricer_it->Midp() * pricer_it->NameMultiplier(), tick_size_, 0);
      pricer_it->SetPriceOffset(curr_tick_offset);
      for (auto [price, qty]: pricer_it->AskOrders()) {
        aggregated_ask_orders_[price + snapshot_ask_tick_offset] += qty;
      }
      for (auto [price, qty]: pricer_it->BidOrders()) {
        aggregated_bid_orders_[price + snapshot_bid_tick_offset] += qty;
      }
    }
  }
  SPDLOG_INFO(
      "aggregated_orders bbo: {}({}) {}({})",
      aggregated_ask_orders_.begin()->first,
      aggregated_ask_orders_.begin()->second,
      aggregated_bid_orders_.begin()->first,
      aggregated_bid_orders_.begin()->second);
  return true;
}

bool RmmStrategy::IsPriceOnlySystemName(const std::string& system_name) {
  return system_name == "Futures.Binance.v1.l1_realtime" ||
         system_name == "Futures.Okex.v5-swap.bbo" ||
         system_name == "Futures.Bybit.v3-linear.bbo";
}

bool RmmStrategy::IsSnapShotOnlySystemName(const std::string& system_name) {
  return system_name.find("realtime_light_bbo") != std::string::npos;
}

void RmmSubStrategy::onTradeFeed(const std::string& product_key, const FeedUpdate& upd) {
  if (price_only_meas_.count(upd.mea()) > 0 ||
      !quote_pos_updated ||
      !maker_fired_) {
    return;
  }
  using coin2::exchange::feed::TradeSide;
  auto curr_ts = upd.timestamp();
  CHECK(feed_keys_.find(product_key) != feed_keys_.end());
  RmmPricer* pricer = nullptr;
  for (auto& pricer_it : pricers_) {
    if (pricer_it->Mea() == upd.mea()) {
      pricer = pricer_it.get();
      break;
    }
  }
  if (!config_ || !pricer || !oe_->HasPosition(*trade_product_holder_)) {
    return;
  }
  if (os()->is_ready() && pricer->BookUpdated() && OrderExecutor()) {
    if (!(pricer->Mea().market == MarketType::Spot) && (!pricer->IndexTimestamp() ||
         pricer->IndexTimestamp() + config_->index_feed_valid_sec() * 1'000'000'000L < curr_ts)) {
      // skip calculation when last ref index feed is not up to date: within [t, t + 60s]
      SPDLOG_WARN("{} Index is not up-to-date!", upd.mea().String());
      return;
    }
    if (!index_ts_ || index_ts_ + config_->index_feed_valid_sec() * 1'000'000'000L < curr_ts) {
      // skip calculation when last prex index feed is not up to date: within [t, t + 60s]
      SPDLOG_WARN("Prex Index is not up-to-date!");
      return;
    }
    if (upd.trade().timestamp + config_->trade_feed_valid_sec() * 1'000'000'000L < curr_ts) {
      // skip when T feed is delayed
      SPDLOG_WARN(
          "T Feed is delayed! ET: {}, FT: {}",
          upd.trade().timestamp,
          curr_ts);
      return;
    }
    SPDLOG_INFO(
        "{} Trade, {} {}({})",
        upd.mea().String(),
        upd.trade().side == TradeSide::TRADE_SELL_SIDE? "Sell" : "Buy",
        upd.trade().price,
        upd.trade().fill_qty);
    if ((pricer->TradeTimestamp() &&
         pricer->TradeTimestamp() + config_->trade_update_period_ns() > curr_ts)) {
      // cumulate T qty during [t, t + 0.5s]
      if (upd.trade().side == TradeSide::TRADE_SELL_SIDE) {
        pricer->SetTradeCumSellQty(pricer->TradeCumSellQty() + upd.trade().fill_qty);
      } else if (upd.trade().side == TradeSide::TRADE_BUY_SIDE) {
        pricer->SetTradeCumBuyQty(pricer->TradeCumBuyQty() + upd.trade().fill_qty);
      }
      return;
    }
    bool reduce_only = strat_num_ == 1 && ((curr_ts - parent_->OtherStratTimestamp()) < 10'000'000'000L);
    pricer->onTradeFeed(upd, reduce_only);
    pricer->SetTradeTimestamp(curr_ts);
  }
}

void RmmSubStrategy::onIndexFeed(const std::string& product_key, const FeedUpdate& upd) {
  CHECK(feed_keys_.find(product_key) != feed_keys_.end());
  auto curr_ts = upd.timestamp();
  if (upd.mea().exchange == ExchangeType::Prex) {
    index_ts_ = upd.timestamp();
    for (auto& pricer_it : pricers_) {
      if (pricer_it->IndexPrice()) {
        pricer_it->FairPriceWindow()->UpdateIndex(curr_ts, upd.index().price, false/* is_ref */);
      }
    }
  } else {
    for (auto& pricer_it : pricers_) {
      if (pricer_it->Mea() == upd.mea()) {
        pricer_it->FairPriceWindow()->UpdateIndex(curr_ts, upd.index().price * pricer_it->NameMultiplier(), true/* is_ref */);
      }
    }
  }

  if (is_sim_) {
    WritePriceToCsv(curr_ts, upd.mea().String(), upd.index().price, false/* is_midp */);
  }
}

void RmmSubStrategy::Hedge(OrderDirection dir, double hedge_amount, int64_t timestamp) {
  if (!hedge_oe_) {
    SPDLOG_WARN("hedge_oe_ is not initialized");
    return;
  }
  double hedge_price;
  if (dir == OrderDirection::DIRECTION_BUY && hedge_book_bid0_.has_value()) {
    hedge_price = hedge_book_bid0_.value() * (1 - config_->hedge_edge_bps() * 1e-4);
    SPDLOG_INFO("buy {} ({}) from binance to hedge position", hedge_amount, hedge_price);
  } else if (dir == OrderDirection::DIRECTION_SELL && hedge_book_ask0_.has_value()) {
    hedge_price = hedge_book_ask0_.value() * (1 + config_->hedge_edge_bps() * 1e-4);
    SPDLOG_INFO("sell {} ({}) from binance to hedge position", hedge_amount, hedge_price);
  } else {
    SPDLOG_WARN("failed to hedge: book empty");
    return;
  }
  const auto oq = hedge_oe_->oq(*hedge_product_holder_);
  auto working_orders = oq->GetAllProductWorkingOrders(hedge_product_->order_native_symbol());
  for (auto* hedge_wo : working_orders) {
    hedge_oe_->CancelOrder(*hedge_product_holder_, timestamp, hedge_wo->proc_order_id());
  }
  hedge_oe_->SubmitOrder(
      *hedge_product_holder_.get(),
      hedge_price,
      hedge_amount,
      dir,
      coin::proto::OrderDuration::GTC_ORDER,
      coin::proto::OrderType::LIMIT_ORDER,
      std::nullopt,
      config_->leverage_rate(),
      true,
      "rmm hedge",
      timestamp,
      true,         // log_fail
      false,        // block_price_band
      std::nullopt, // maxpos_multiplier
      std::nullopt, // reduce_only
      false);       // force_submit
}

void RmmSubStrategy::onAccountInfo(const OrderUpdate& upd) {
  auto& account_info = upd.account_info();
  if (!account_info.is_ready()) {
    SPDLOG_INFO("account info not ready");
    return;
  }
  const auto timestamp = upd.timestamp();
  const auto& prod = *trade_product_;
  auto account_type = GetAccountType(account_info);
  if (account_type == AccountType::RMM_STRAT_0 && account_info.HasPosition(prod)) {
    strat0_position_ = std::make_pair(account_info.GetPosition(prod).net_position(), timestamp);
    SPDLOG_INFO("strat0's position: {}", account_info.GetPosition(prod).net_position());
  } else if (account_type == AccountType::RMM_STRAT_1 && account_info.HasPosition(prod)) {
    strat1_position_ = std::make_pair(account_info.GetPosition(prod).net_position(), timestamp);
    SPDLOG_INFO("strat1's position: {}", account_info.GetPosition(prod).net_position());
  } else if (account_type == AccountType::HEDGE && config_const_.enable_hedge() &&
    account_info.HasPosition(*hedge_product_)) {
    hedge_position_ = std::make_pair(account_info.GetPosition(*hedge_product_).net_position(), timestamp);
    SPDLOG_INFO("hedge position: {}", account_info.GetPosition(*hedge_product_).net_position());
  } else {
    SPDLOG_WARN("Account info from unknown account");
    return;
  }
  auto strat_0_pos = GetPosition(AccountType::RMM_STRAT_0, timestamp);
  auto strat_1_pos = GetPosition(AccountType::RMM_STRAT_1, timestamp);
  auto hedge_pos = GetPosition(AccountType::HEDGE, timestamp);
  if (strat_0_pos.has_value() && strat_1_pos.has_value()) {
    auto rmm_position = strat_0_pos.value() + strat_1_pos.value();
    rmm_position_window_.Update(upd.timestamp(), rmm_position);
    SPDLOG_INFO("rmm_position_window_'s mean: {}, stddev: {}",
      rmm_position_window_.average(), rmm_position_window_.simplestd());
    if (hedge_pos.has_value()) {
      SPDLOG_INFO("rmm total position: {}, hedge position: {}, total: {}",
        rmm_position, hedge_pos.value(), rmm_position + hedge_pos.value());
    } else {
      SPDLOG_INFO("rmm total position: {}", rmm_position);
    }
  }
  if (hedge_pos.has_value() &&
      last_hedge_ts_ + config_->hedge_period_sec() * 1'000'000'000L < upd.timestamp() &&
      rmm_position_window_.num_elements() > config_->hedge_min_rmm_position_num()) {
    // TODO(peter) : refactor this
    double rmm_position = rmm_position_window_.average() -
      std::clamp(rmm_position_window_.average(), -hedge_exception_qty, hedge_exception_qty);
    double total_position = hedge_pos.value() + rmm_position;
    OrderDirection direction;
    double hedge_amount;
    if (total_position > 0) {
      direction = OrderDirection::DIRECTION_SELL;
      hedge_amount = std::max(total_position
        - rmm_position_window_.simplestd() * config_->hedge_stddev_multiplier(), 0.0) / 2;
    } else {
      direction = OrderDirection::DIRECTION_BUY;
      hedge_amount = std::max(-total_position
        - rmm_position_window_.simplestd() * config_->hedge_stddev_multiplier(), 0.0) / 2;
    }
    if (hedge_amount > 0 && config_->enable_hedge() && hedge_os()->is_ready()) {
      Hedge(direction, hedge_amount, upd.timestamp());
      last_hedge_ts_ = upd.timestamp();
    }
  }
}


void RmmSubStrategy::onAccountOrder(const OrderUpdate& upd) {
  if (!oe_ || !config_.has_value() || !trade_product_holder_) {
    return;
  }
  if (upd.product_order_info().event().type() == OrderEvent::ORDER_AUTO_CANCELED) {
    auto& oevent = upd.product_order_info().event();
    auto odir = OrderDirection::DIRECTION_BUY;
    if (oevent.trade_side() == TradeSide::TRADE_BUY_SIDE) {
    } else if (oevent.trade_side() == TradeSide::TRADE_SELL_SIDE) {
      odir = OrderDirection::DIRECTION_SELL;
    } else {
      NOTREACHED() << oevent.trade_side();
    }
    if (oevent.tag() != "retry" && oevent.order_price() > 0) {
      if (oevent.order_type() != coin::proto::OrderType::IOC_LIMIT_ORDER) {
        oe_->SubmitOrder(
            *trade_product_holder_,
            oevent.order_price(),
            oevent.order_qty(),
            odir,
            coin::proto::OrderDuration::GTC_ORDER,
            coin::proto::OrderType::LIMIT_ORDER,
            std::nullopt,
            config_->leverage_rate(),
            true,
            "retry",
            upd.timestamp());
      }
    }
  }
}

RmmPricer::RmmPricer(
    RmmSubStrategy* parent,
    const IProduct* product,
    IProductHolder* pholder,
    std::string mea_string,
    bool is_sim)
  : is_sim_(is_sim),
    parent_(parent),
    fair_price_window_(parent_->ConfigConst().index_price_window_ns()),
    spread_window_(30'000'000'000L),
    fair_price_offset_window_(600'000'000'000L) {
  ref_product_holder_.reset(pholder);
  mea_ = MarketExchangeApi::FromString(mea_string);
  if (mea_.market == MarketType::Spot) {
    spot_ref_ = true;
  } else if (mea_.market == MarketType::Futures) {
    spot_ref_ = false;
  } else {
    CHECK(false) << "Incompatible MarketType: " << MarketTypeToString(mea_.market);
  }
  multiplier_ = mea_.market == MarketType::Spot ?
    1.0 :
    ref_product_holder_->product_info().contract_value();
  if (parent_->ConfigConst().ref_weights().find(mea_string) != parent_->ConfigConst().ref_weights().end()) {
    ref_weight_ = (parent_->ConfigConst().ref_weights()).at(mea_string);
  }
  GetNameMultiplier(product->absolute_norm(), name_multiplier_);
  quote_str_ = product->quote().currency();
}


bool RmmPricer::onBookFeed(const FeedUpdate& upd) {
  if (HasMidPrice(upd)) {
    SPDLOG_INFO(
        "{} Book, bbo {}({}) {}({})",
        upd.mea().String(),
        upd.book().Ask0().value().price,
        upd.book().Ask0().value().qty,
        upd.book().Bid0().value().price,
        upd.book().Bid0().value().qty);
    if (book_updated_ &&
        ask0p_ == upd.book().Ask0().value().price &&
        bid0p_ == upd.book().Bid0().value().price &&
        ask0q_ == upd.book().Ask0().value().qty &&
        bid0q_ == upd.book().Bid0().value().qty) {
      return false;
    }
    int64_t curr_ts = upd.book().Timestamp() ? upd.book().Timestamp() : upd.timestamp();
    book_updated_ = true;
    ask0p_ = upd.book().Ask0().value().price;
    bid0p_ = upd.book().Bid0().value().price;
    ask0q_ = upd.book().Ask0().value().qty;
    bid0q_ = upd.book().Bid0().value().qty;
    tick_size_ = ref_product_holder_->product_info().tick_price().UpTicksize(ask0p_);
    spread_window_.PushElement(
        curr_ts, upd.book().Ask0().value().price - upd.book().Bid0().value().price);
    spread_window_.UpdateTimeOnly(curr_ts);
    if (is_sim_) {
      parent_->WritePriceToCsv(curr_ts, upd.mea().String(), 0.5 * (ask0p_ + bid0p_), true/* is_midp */);
    }
    SPDLOG_INFO_EVERY_NS(
        10'000'000'000L,
        curr_ts,
        "{} Spread ma: {}",
        upd.mea().String(), spread_window_.average());
    return true;
  } else {
    SPDLOG_WARN("NO MIDP: {} feed", upd.product().absolute_norm());
    return false;
  }
}

template <typename T1, typename T2>
void RmmPricer::UpdateSnapshotFeed(
    const T1& feed_orders, T2* side_orders, int32_t sign) {
  for (auto& pq : feed_orders) {
    if (pq.second == 0) {
      continue;
    }
    auto int_price_parent = ToIntKey(pq.first * *name_multiplier_, parent_->tick_size(), sign);
    int64_t price_jitter = 0;
    if (parent_->tick_size() < tick_size_) {
      // if prex tick size is smaller (denser), add randomized jitter to price
      int64_t modular = static_cast<int64_t>(tick_size_ / parent_->tick_size());
      price_jitter = std::rand() % modular;
    }
    (*side_orders)[int_price_parent - sign * price_jitter] +=
      pq.second / *name_multiplier_ * multiplier_;
  }
}

void RmmPricer::UpdateRefOrders(const FeedUpdate& upd, int64_t timestamp) {
  CHECK(upd.book().Ask0().has_value() && upd.book().Bid0().has_value());
  if (timestamp - book_reset_ts_ > 5L * 1'000'000'000L) {
    // do not process delta book feeds and process snapshot book (for realtime_light_bbo recipe)
    // also, we do not process every snapshot book feed since it is quite heavy
    ask_orders_.clear();
    bid_orders_.clear();
    book_reset_ts_ = timestamp;
    double snapshot_ask0p_ = upd.book().Ask0().value().price;
    double snapshot_bid0p_ = upd.book().Bid0().value().price;
    snapshot_midp_ = 0.5 * (snapshot_ask0p_ + snapshot_bid0p_);

    UpdateSnapshotFeed<>(upd.book().Asks(), &ask_orders_, -1);
    UpdateSnapshotFeed<>(upd.book().Bids(), &bid_orders_, 1);
  }
  auto ask0_int_price = ToIntKey(upd.book().Ask0().value().price, tick_size_, -1);
  auto bid0_int_price = ToIntKey(upd.book().Bid0().value().price, tick_size_, 1);
  auto ask0_qty = upd.book().Ask0().value().qty;
  auto bid0_qty = upd.book().Bid0().value().qty;
  SPDLOG_INFO_EVERY_NS(
      10'000'000'000L,
      timestamp,
      "{} bbo(prex): {}({}), {}({}), bbo(raw): {}({}), {}({})",
      upd.mea().String(),
      ask_orders_.begin()->first,
      ask_orders_.begin()->second,
      bid_orders_.begin()->first,
      bid_orders_.begin()->second,
      ask0_int_price,
      ask0_qty,
      bid0_int_price,
      bid0_qty);
  if ((ask0_int_price <= bid0_int_price) ||
      (!ask_orders_.empty() && !bid_orders_.empty() &&
       ask_orders_.begin()->first <= bid_orders_.begin()->first)) {
    SPDLOG_ERROR("Book Crosses");
  }
  book_ts_ = timestamp;
  SPDLOG_INFO_EVERY_NS(
      10'000'000'000L,
      timestamp,
      "{} Book feed size: ask {}, bid {}",
      upd.mea().String(), ask_orders_.size(), bid_orders_.size());
}

void RmmSubStrategy::UpdateWorkingOrders(int64_t timestamp) {
  working_ask_orders_.clear();
  working_bid_orders_.clear();
  ask_ocs_.clear();
  bid_ocs_.clear();

  int64_t order_lifetime_ns = config_->oe_config().common().cancel().order_lifetime_ns();
  const auto oq = oe_->oq(*trade_product_holder_);
  auto working_orders = oq->GetAllProductWorkingOrders(trade_product_->order_native_symbol());
  SPDLOG_INFO(oe_->DebugString(timestamp));
  for (auto* order : working_orders) {
    if (order->has_exchange_order_id() == false &&
      order->order_created_time() + order_lifetime_ns * 2 < timestamp) {
      SPDLOG_INFO("zombie found, resubmit cancel, cancel created {}, id is {}", order->cancel_created_time(), order->proc_order_id());
      oe_->CancelOrder(*trade_product_holder_, timestamp, order->proc_order_id());
      continue;
    }
    if ((ask_ocs_.size() >= 1 && order->direction() == OrderDirection::DIRECTION_SELL) ||
        (bid_ocs_.size() >= 1 && order->direction() == OrderDirection::DIRECTION_BUY)) {
      if (order->GetAmendedCnt() > 10'000) {
        SPDLOG_INFO("amended for {} times, id is {}", order->GetAmendedCnt(), order->proc_order_id());
        oe_->CancelOrder(*trade_product_holder_, timestamp, order->proc_order_id());
        continue;
      } else if (order->GetAmendedCnt() * order->filled_cnt() > 1'000'000) {
        SPDLOG_INFO("amended for {} times, filled {} times, id is {}",
                      order->GetAmendedCnt(), order->filled_cnt(), order->proc_order_id());
        oe_->CancelOrder(*trade_product_holder_, timestamp, order->proc_order_id());
        continue;
      }
    }
    if (order->IsCancelSent()) {
      SPDLOG_INFO("cancel sent order, id is {}", order->cancel_created_time(), order->proc_order_id());
      oe_->CancelOrder(*trade_product_holder_, timestamp, order->proc_order_id());
      continue;
    }
    if (order->filled_qty() > order->working_qty()) {
      oe_->CancelOrder(*trade_product_holder_, timestamp, order->proc_order_id());
      continue;
    }
    if (order->direction() == OrderDirection::DIRECTION_SELL) {
      ask_ocs_.push_back(order);
    } else if (order->direction() == OrderDirection::DIRECTION_BUY) {
      bid_ocs_.push_back(order);
    }
  }

  std::sort(ask_ocs_.begin(), ask_ocs_.end(), [](const auto* wo1, const auto* wo2) {
    return wo1->order_price() < wo2->order_price();
  });
  std::sort(bid_ocs_.begin(), bid_ocs_.end(), [](const auto* wo1, const auto* wo2) {
    return wo1->order_price() > wo2->order_price();
  });

  if (strat_num_ == 0) {
    num_order_ = config_->num_bot0_order();
  } else {
    num_order_ = outside_only_ ? config_->num_bot1_order() : config_->num_bot0_order();
  }
  if (ask_ocs_.size() != num_order_ || bid_ocs_.size() != num_order_) {
    SPDLOG_INFO("[amend] strange wo size (wo sizes: {} {})", ask_ocs_.size(), bid_ocs_.size());
  }

  if (last_update_ts_ > 0) {
    ref_downtime_window_.Update(timestamp, std::max(0L, timestamp - last_update_ts_ - config_->book_update_period_ns()));
  }
  SPDLOG_INFO("ref downtime sec: {}", (double)(ref_downtime_window_.sum()) / 1e9);
  last_update_ts_ = timestamp;
}

void RmmSubStrategy::FireMakerOrders(int64_t order_id,
                                     double price,
                                     OrderDirection direction,
                                     int32_t sign,
                                     int64_t timestamp) {
  double mispricebp_ma = 0;
  if (mispricebp_window_.num_elements() >= 30) {
    mispricebp_ma = mispricebp_window_.average();
  }
  const double takerthres_base = config_->taker_threshold_bps();
  // if avg skew is 4 bp, then exp(-1) = 0.36
  double buytakerthres = takerthres_base;
  double selltakerthres = takerthres_base;
  if (mispricebp_ma < 0) {
    buytakerthres *= exp(-(std::abs(mispricebp_ma) * 0.25));
  } else {
    selltakerthres *= exp(-(std::abs(mispricebp_ma) * 0.25));
  }
  SPDLOG_INFO(
      "TAKERTHRES = B[{}]/S[{}] : MISPRICEBP_MA: [{}]",
      buytakerthres, selltakerthres, mispricebp_ma);
  double takerthres = (sign == 1) ? buytakerthres : selltakerthres;

  double fire_qty = config_->const_book_depth_quote() / price;
  if (config_->order_qty_ub() != 0) {
    fire_qty = std::min(fire_qty, config_->order_qty_ub());
  }

  if (price > 0 && fair_midp_ > 0 &&
      config_->oe_config().common().qty().min_order_qty() > 0 &&
      fire_qty >= config_->oe_config().common().qty().min_order_qty()) {
    // use taker order for outer prices
    double advantage_bps = std::abs(price / fair_midp_ - 1) * 1e4;
    bool post_only = advantage_bps < takerthres || config_->maker_only();
    SPDLOG_INFO("SUB[{}] {} @ {}", sign, fire_qty, price);
    if (order_id == 0) {
      oe_->SubmitOrder(
          *trade_product_holder_,
          price,
          fire_qty,
          direction,
          coin::proto::OrderDuration::GTC_ORDER,
          coin::proto::OrderType::LIMIT_ORDER,
          std::nullopt,
          config_->leverage_rate(),
          post_only,
          fmt::format("rmm maker {}", config_->modular_remainder()),
          timestamp,
          true,         // log_fail
          false,        // block_price_band
          std::nullopt, // maxpos_multiplier
          std::nullopt, // reduce_only
          false);       // force_submit
    } else {
      oe_->AmendOrder(
          *trade_product_holder_,
          order_id,
          price,
          fire_qty * qty_randomizer.rand(),
          direction,
          post_only,
          timestamp,
          true,         // log_fail
          std::nullopt, // maxpos_multiplier
          std::nullopt, // reduce_only
          false,
          fire_qty);       // force_submit
    }
  }
}

void RmmPricer::onTradeFeed(const FeedUpdate& upd, bool reduce_only) {
  return;
  using coin2::exchange::feed::TradeSide;
  int64_t curr_ts = upd.timestamp();
  if (static_cast<int64_t>(upd.trade().timestamp / 1e9) % parent_->Config().modular_divisor() !=
      parent_->Config().modular_remainder()) {
    return;
  }
  double price = upd.trade().price * *name_multiplier_ + price_offset_ * parent_->tick_size();
  double qty = upd.trade().fill_qty;
  double volume_multiplier = parent_->Config().trade_volume_multiplier();
  if (upd.trade().side == TradeSide::TRADE_SELL_SIDE) {
    qty += trade_cum_sell_qty_;
    if (qty / *name_multiplier_* multiplier_ * volume_multiplier < parent_->tick_qty()) {
      // if submit qty < 1 tick_qty, then keep cumulating qty
      trade_cum_sell_qty_ = qty;
      return;
    }
    trade_cum_sell_qty_ = 0;
  } else if (upd.trade().side == TradeSide::TRADE_BUY_SIDE) {
    qty += trade_cum_buy_qty_;
    if (qty / *name_multiplier_* multiplier_ * volume_multiplier < parent_->tick_qty()) {
      // if submit qty < 1 tick_qty, then keep cumulating qty
      trade_cum_buy_qty_ = qty;
      return;
    }
    trade_cum_buy_qty_ = 0;
  }
  double submittable_min_qty =
    std::max(parent_->trade_product_holder_->product_info().tick_qty().submittable_min_qty(),
             parent_->trade_product_holder_->product_info().min_amount() / price);
  if (price > 0 && qty > 0) {
    parent_->OrderExecutor()->SubmitOrder(
        *parent_->TradeProductHolder(),
        price,
        submittable_min_qty,
        // qty / *name_multiplier_* multiplier_ * volume_multiplier,
        upd.trade().side == TradeSide::TRADE_BUY_SIDE?
        OrderDirection::DIRECTION_BUY : OrderDirection::DIRECTION_SELL,
        coin::proto::OrderDuration::IOC_ORDER,
        coin::proto::OrderType::IOC_LIMIT_ORDER,
        std::nullopt,
        parent_->Config().leverage_rate(),
        false,
        fmt::format("rmm taker {}", parent_->Config().modular_remainder()),
        curr_ts,
        true,         // log_fail
        false,        // block_price_band
        std::nullopt, // maxpos_multiplier
        reduce_only,
        false);       // force_submit
  }
}

}  // namespace coin2::strategy::rmm_strat
